package org.example.rabbitmq.publisherconfirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.function.BooleanSupplier;

public class PublisherConfirms {

    static final int MESSAGE_COUNT = 50_000;

    static Connection createConnection() throws Exception {
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("192.168.0.228");
        cf.setUsername("mq");
        cf.setPassword("123456");
        return cf.newConnection();
    }

    public static void main(String[] args) throws Exception {
        publishMessagesIndividually();
        publishMessagesInBatch();
        handlePublishConfirmsAsynchronously();
    }

    /**
     * 单独发布消息到RabbitMQ队列中
     * 此方法演示如何创建一个临时队列，并单独发布消息到该队列中
     * 每条消息在发送后都会等待确认，以确保消息成功到达队列
     * 此外，该方法还会输出发布消息的总时间和消息发布速率
     *
     * @throws Exception 如果在创建连接、通道或发布消息过程中发生错误，则抛出此异常
     */
    static void publishMessagesIndividually() throws Exception {
        // 创建并打开与RabbitMQ的连接
        try (Connection connection = createConnection()) {
            // 通过连接创建一个通道
            Channel ch = connection.createChannel();

            // 生成一个唯一的队列名称，并声明一个临时队列
            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);

            // 将通道设置为确认模式，以确保每条消息都已被接收和确认
            ch.confirmSelect();

            // 记录开始时间，用于计算发布消息所需的总时间
            long start = System.nanoTime();

            // 循环发布消息到队列中
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                // 消息体内容为当前循环的索引值
                String body = String.valueOf(i);
                // 发布消息到指定队列
                ch.basicPublish("", queue, null, body.getBytes());
                // 等待消息确认，如果5秒内未确认，则抛出异常
                ch.waitForConfirmsOrDie(5_000);
            }

            // 记录结束时间，用于计算发布消息所需的总时间
            long end = System.nanoTime();

            // 输出发布消息的总数量和所花费的时间
            System.out.format("Published %,d messages individually in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis());
        }
    }


    /**
     * 以批处理方式发布消息到队列中
     * 此方法创建一个临时队列，并以批处理方式发布指定数量的消息到该队列中
     * 使用确认模式来确保消息被成功发布，以提高消息发送的可靠性和效率
     *
     * @throws Exception 如果消息发布过程中发生错误
     */
    static void publishMessagesInBatch() throws Exception {
        // 创建到RabbitMQ服务器的连接
        try (Connection connection = createConnection()) {
            // 创建一个通道
            Channel ch = connection.createChannel();

            // 生成一个唯一的队列名称，并声明一个临时队列
            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);

            // 将通道设置为确认模式
            ch.confirmSelect();

            // 定义批处理大小和当前未确认的消息计数
            int batchSize = 100;
            int outstandingMessageCount = 0;

            // 记录开始时间，用于计算处理时间
            long start = System.nanoTime();
            // 循环发布消息
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                // 消息体内容
                String body = String.valueOf(i);
                // 发布消息到队列中
                ch.basicPublish("", queue, null, body.getBytes());
                // 增加未确认的消息计数
                outstandingMessageCount++;

                // 如果达到批处理大小，则等待所有消息被确认
                if (outstandingMessageCount == batchSize) {
                    ch.waitForConfirmsOrDie(5_000);
                    // 重置未确认的消息计数
                    outstandingMessageCount = 0;
                }
            }

            // 如果还有未确认的消息，则等待它们被确认
            if (outstandingMessageCount > 0) {
                ch.waitForConfirmsOrDie(5_000);
            }
            // 记录结束时间
            long end = System.nanoTime();
            // 输出消息发布统计信息
            System.out.format("Published %,d messages in batch in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis());
        }
    }


    /**
     * 异步处理发布确认
     * 该方法用于异步处理消息发布后的确认，确保消息成功发送到RabbitMQ服务器
     * 它通过创建一个临时队列并监听确认事件来实现
     *
     * @throws Exception 如果在处理过程中遇到错误，则抛出异常
     */
    static void handlePublishConfirmsAsynchronously() throws Exception {
        // 创建到RabbitMQ的连接
        try (Connection connection = createConnection()) {
            Channel ch = connection.createChannel();

            // 生成一个唯一的队列名称并声明一个临时队列
            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);

            // 启用发布确认模式
            ch.confirmSelect();

            // 存储未确认的消息
            ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

            // 定义处理确认消息的回调
            ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {
                if (multiple) {
                    // 如果是批量确认，清除所有小于等于当前序列号的消息
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(
                            sequenceNumber, true
                    );
                    confirmed.clear();
                } else {
                    // 否则，只清除当前序列号对应的消息
                    outstandingConfirms.remove(sequenceNumber);
                }
            };

            // 添加确认监听器，同时处理确认和未确认的消息
            ch.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> {
                String body = outstandingConfirms.get(sequenceNumber);
                System.err.format(
                        "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
                        body, sequenceNumber, multiple
                );
                cleanOutstandingConfirms.handle(sequenceNumber, multiple);
            });

            // 开始发布消息
            long start = System.nanoTime();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String body = String.valueOf(i);
                outstandingConfirms.put(ch.getNextPublishSeqNo(), body);
                ch.basicPublish("", queue, null, body.getBytes());
            }

            // 等待直到所有消息都被确认
            if (!waitUntil(Duration.ofSeconds(60), () -> outstandingConfirms.isEmpty())) {
                throw new IllegalStateException("All messages could not be confirmed in 60 seconds");
            }

            long end = System.nanoTime();
            // 输出消息发布和处理确认的时间
            System.out.format("Published %,d messages and handled confirms asynchronously in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis());
        }
    }


    /**
     * 等待直到某个条件在指定时间内变为真
     * 此方法用于在不超过给定时间限制的情况下，重复检查一个条件，直到该条件为真
     * 它可以帮助在并发编程或需要等待特定状态时，避免无限循环或过早退出
     *
     * @param timeout   一个Duration对象，表示等待条件变为真的最大时间
     * @param condition 一个BooleanSupplier对象，表示需要评估的条件
     * @return 如果条件在指定时间内变为真，则返回true；否则返回false
     * @throws InterruptedException 如果线程在等待期间被中断
     */
    static boolean waitUntil(Duration timeout, BooleanSupplier condition) throws InterruptedException {
        // 初始化已等待的时间为0毫秒
        int waited = 0;

        // 当条件不为真且已等待的时间未超过总超时时间时，继续等待
        while (!condition.getAsBoolean() && waited < timeout.toMillis()) {
            // 线程休眠100毫秒
            Thread.sleep(100L);
            // 更新已等待的时间
            waited += 100;
        }

        // 返回最终的条件值，无论是否超时
        return condition.getAsBoolean();
    }


}
